package com.deepclassifieds.daas.api.agg;

import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.UUID;

import com.deepclassifieds.daas.api.DaasAggService;
import com.deepclassifieds.daas.api.DaasExpression;
import com.deepclassifieds.daas.api.DaasExpressionBuilder;
import com.deepclassifieds.daas.api.DaasShardService;
import com.deepclassifieds.daas.api.DaasUID;
import com.deepclassifieds.daas.api.agg.name.AggDatabaseName;
import com.deepclassifieds.daas.api.agg.name.AggTableName;
import com.deepclassifieds.daas.api.builder.ColumnDefsBuilder;
import com.deepclassifieds.daas.api.builder.ColumnsBuilder;
import com.deepclassifieds.daas.api.builder.RecordBuilder;
import com.deepclassifieds.daas.api.builder.ShardsBuilder;
import com.deepclassifieds.daas.api.jdbc.DaasExpressionBuilderJdbcImpl;
import com.deepclassifieds.daas.api.jdbc.DaasShardServiceJdbcImpl;
import com.deepclassifieds.daas.api.jdbc.dialect.Dialect;
import com.deepclassifieds.daas.api.jdbc.dialect.DialectHSQLDBImpl;
import com.deepclassifieds.daas.api.jdbc.sql.BoneCPConnectionProvider;
import com.deepclassifieds.daas.api.xsd.ColumnDef;
import com.deepclassifieds.daas.api.xsd.ColumnDefs;
import com.deepclassifieds.daas.api.xsd.Database;
import com.deepclassifieds.daas.api.xsd.Databases;
import com.deepclassifieds.daas.api.xsd.Datatype;
import com.deepclassifieds.daas.api.xsd.Record;
import com.deepclassifieds.daas.api.xsd.Records;
import com.deepclassifieds.daas.api.xsd.Shard;
import com.deepclassifieds.daas.api.xsd.Shards;
import com.deepclassifieds.daas.api.xsd.Table;
import com.deepclassifieds.daas.api.xsd.Tables;
import com.deepclassifieds.daas.api.xsd.impl.ShardImpl;
import com.deepclassifieds.daas.common.Name;
import com.deepclassifieds.daas.common.Reject;
import com.deepclassifieds.daas.common.sql.ConnectionProvider;

public class DaasAggServiceImpl implements DaasAggService {
	
	private static final String JDBC_DRIVER      = "org.hsqldb.jdbc.JDBCDriver";
	private static final String JDBC_URL         = "jdbc:hsqldb:mem:daas_service_agg";
	private static final String JDBC_USERNAME    = "sa";
	private static final String JDBC_PASSWORD    = "";
	
	private static final Name DAAS_AGG_SHARD_NUMBER   = new Name("daas_agg_shard_number");
	private static final Random random = new Random(); 
	
	/** List of virtual shards */
	private final Shards vitrualShards = new Shards();
	/** List of Daas Service JDBC implementations mappings for each virtual shard */
	private final List<DaasShardServiceJdbcImpl> daasShardServices;
	/** Map of physical Daas Services */
	private final Map<String,DaasShardServiceJdbcImpl> daasServiceMap;
	/** Single Shard Aggregator in-memory Daas Service */
	private final DaasShardService aggDaasService;
	/** Single Shard for Aggregator Service */
	private final Shard aggShard;
	
	public DaasAggServiceImpl(
			Dialect dialect,
			List<ConnectionProvider> connectionProvidersToShards) {
		Reject.ifNull(dialect);
		Reject.ifNull(connectionProvidersToShards);
		Reject.ifFalse(connectionProvidersToShards.size() > 0);
		int size = connectionProvidersToShards.size();
		for(int i = 0; i < size; ++i) {
			ConnectionProvider provider = connectionProvidersToShards.get(i);
			ShardImpl shard = new ShardImpl();
			shard.setNumber(i);
			shard.setValue(provider.getUniqueIdentifier());
			vitrualShards.getShard().add(shard);
		}
		this.daasShardServices = new ArrayList<DaasShardServiceJdbcImpl>(size);
		this.daasServiceMap = new LinkedHashMap<String,DaasShardServiceJdbcImpl>(size);
		for(int i = 0; i < size; ++i) {
			ConnectionProvider provider = connectionProvidersToShards.get(i);
			DaasShardServiceJdbcImpl impl = daasServiceMap.get(provider.getUniqueIdentifier());
			if(impl == null) {
				impl = new DaasShardServiceJdbcImpl(vitrualShards, dialect, provider);
				daasServiceMap.put(provider.getUniqueIdentifier(), impl);
			}
			daasShardServices.add(impl);
		}
		//
		// crate aggregator daas service
		//
		this.aggShard = new ShardsBuilder().createAndAddShard(0); 
		this.aggDaasService = new DaasShardServiceJdbcImpl(
				new Shards(), new DialectHSQLDBImpl(), 
				new BoneCPConnectionProvider(JDBC_DRIVER, JDBC_URL, JDBC_USERNAME, JDBC_PASSWORD));
	}

	@Override
	public Databases getDatabases() {
		throw new RuntimeException("not implemented yet...");
	}

	@Override
	public Database getDatabaseByName(Name databaseName) {
		throw new RuntimeException("not implemented yet...");
	}

	@Override
	public Shards getShards() {
		return vitrualShards;
	}

	@Override
	public Shard getShardByNumber(int shardNumber) {
		return vitrualShards.getShard().get(shardNumber);
	}

	@Override
	public Tables getTables(Database database) {
		throw new RuntimeException("not implemented yet...");

	}

	@Override
	public Table getTableByName(Database database, Shard shard, Name tableName) {
		throw new RuntimeException("not implemented yet...");
	}

	@Override
	public Database createDatabase(final Name databaseName) {
		return applyAll(new Apply<Database>() {
			@Override
			public Database apply(Shard shard, DaasShardServiceJdbcImpl daasShardService) {
				return daasShardService.createDatabase(databaseName, shard);
			}
		});
	}
	
	@Override
	public void dropDatabase(final Database database) {
		applyAll(new Apply<Void>() {
			@Override
			public Void apply(Shard shard, DaasShardServiceJdbcImpl daasService) {
				daasService.dropDatabase(database, shard);
				return null;
			}
		});
	}

	@Override
	public Table createTable(final Database database, final Name tableName, final ColumnDefs columnDefs) {
		return applyAll(new Apply<Table>() {
			@Override
			public Table apply(Shard shard, DaasShardServiceJdbcImpl daasService) {
				return daasService.createTable(database, shard, tableName, columnDefs);
			}
		});
	}

	@Override
	public void dropTable(final Database database, final Table table) {
		applyAll(new Apply<Void>() {
			@Override
			public Void apply(Shard shard, DaasShardServiceJdbcImpl daasService) {
				daasService.dropTable(database, shard, table);
				return null;
			}
		});
	}

	@Override
	public void alterTableAddColumn(final Database database, final Table table, final ColumnDef columnDef) {
		applyAll(new Apply<Void>() {
			@Override
			public Void apply(Shard shard, DaasShardServiceJdbcImpl daasShardService) {
				daasShardService.alterTableAddColumn(database, shard, table, columnDef);
				return null;
			}
		});
	}

	@Override
	public void alterTableDropColumn(final Database database, final Table table, final Name columnName) {
		applyAll(new Apply<Void>() {
			@Override
			public Void apply(Shard shard, DaasShardServiceJdbcImpl daasShardService) {
				daasShardService.alterTableDropColumn(database, shard, table, columnName);
				return null;
			}
		});
	}

	@Override
	public DaasExpressionBuilder createExpressionBuilder() {
		return new DaasExpressionBuilderJdbcImpl();
	}

	@Override
	public Record insertRecord(Database database, Shard shard, Table table, Record record) {
		Reject.ifNull(database);
		Reject.ifNull(shard);
		Reject.ifNull(table);
		Reject.ifNull(record);
		DaasShardServiceJdbcImpl daasService = daasShardServices.get(shard.getNumber());
		return daasService.insertRecord(database, shard, table, record);
	}

	@Override
	public Record selectRecordByUID(Database database, Table table, DaasUID daasUID) {
		throw new RuntimeException("not implemented yet...");
	}

	@Override
	public Record selectRecordByUUID(Database database, Shards shards, Table table, UUID uuid) {
		throw new RuntimeException("not implemented yet...");
	}

	@Override
	public Records selectRecords(Database database, Shards shards, Table table, DaasExpression where) {
		Reject.ifNull(database);
		Reject.ifNull(shards);
		Reject.ifNull(table);
		Reject.ifNull(where);
		if(shards.getShard().size() == 0) return new Records();
		if(shards.getShard().size() == 1) {
			DaasShardServiceJdbcImpl daasService = daasShardServices.get(shards.getShard().get(0).getNumber());
			return daasService.selectRecords(database, shards.getShard().get(0), table, where);
		} else {
			DaasShardServiceJdbcImpl daasServiceMetadata = daasShardServices.get(shards.getShard().get(0).getNumber());
			ColumnDefs columnDefs = daasServiceMetadata.getTableByName(
					database, shards.getShard().get(0), new Name(table.getName())).getColumnDefs();
			int random = nextRandomPositiveInt();
			//
			// create temporary database
			//
			Database aggDatabase = aggDaasService.createDatabase(
					new AggDatabaseName(new Name(database.getName()), random).build(), aggShard);
			//
			// create temporary table in aggregator database
			//
			ColumnDefsBuilder columnDefsBuilder = new ColumnDefsBuilder(columnDefs);
			columnDefsBuilder.createAndAddColumnDef(DAAS_AGG_SHARD_NUMBER, Datatype.INTEGER);
			Table aggTable = aggDaasService.createTable(aggDatabase, aggShard, 
					new AggTableName(new Name(table.getName()), random).build(), columnDefsBuilder.build());
			//
			// sequential for now, will be rewritten in parallel later
			//
			for(Shard shard : shards.getShard()) {
				DaasShardServiceJdbcImpl daasService = daasShardServices.get(shard.getNumber());
				Records records = daasService.selectRecords(database, shard, table, where);
				for(Record record : records.getRecord()) {
					//
					// append DAAS_AGG_SHARD_NUMBER column and insert into aggregator table
					//
					ColumnsBuilder columnsBuilder = new ColumnsBuilder(record.getColumns());
					columnsBuilder.createAndAddColumn(DAAS_AGG_SHARD_NUMBER, Integer.toString(shard.getNumber()));
					RecordBuilder recordBuilder = new RecordBuilder();
					recordBuilder.addColumns(columnsBuilder.build());
					aggDaasService.insertRecord(aggDatabase, aggShard, aggTable, recordBuilder.build());
				}
			}
			//
			// run query against aggregator
			//
			Records records = aggDaasService.selectRecords(aggDatabase, aggShard, aggTable, where);
			//
			// drop temporary table in aggregator database
			//
			aggDaasService.dropTable(aggDatabase, aggShard, aggTable);
			//
			// drop temporary database
			//
			aggDaasService.dropDatabase(aggDatabase, aggShard);
			//
			// return result
			//
			return records;
		}
	}

	@Override
	public void updateRecord(Database database, Shard shard, Table table,
			Record record) {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void deleteRecordByUID(Database database, Table table,
			DaasUID daasUID) {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void deleteRecordByUUID(Database database, Shards shards,
			Table table, UUID uuid) {
		// TODO Auto-generated method stub
		
	}

	// ----------------------------------------------------------------------------------------------------------------
	//   helper methods below
	// ----------------------------------------------------------------------------------------------------------------
	
	private interface Apply<T> {
		T apply(Shard shard, DaasShardServiceJdbcImpl daasService);
	}
	
	/**
	 * Apply operation to all physical shards closure like.
	 * @param <T> Type to apply to.
	 * @param apply Single operation apply interface.
	 * @return Result of type <T>.
	 */
	private <T> T applyAll(Apply<T> apply) {
		T result = null;
		int size = vitrualShards.getShard().size();
		for(int i = 0; i < size; ++i) {
			Shard shard = vitrualShards.getShard().get(i);
			DaasShardServiceJdbcImpl daasService = daasShardServices.get(i);
			result = apply.apply(shard, daasService);
			
		}
//		for(Map.Entry<String,DaasServiceJdbcImpl> entry : daasServiceMap.entrySet()) {
//			result = apply.apply(entry.getValue());
//		}
		return result;
	}
	
	private int nextRandomPositiveInt() {
		return random.nextInt(Integer.MAX_VALUE);
	}

	
}